/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZMsg;

public class CSBWorker implements Runnable { 
    private static Logger LOG = Logger.getLogger(CSBWorker.class.getName());

    private String conn;
    private String id;
    private ZContext ctx;
    
    private Socket worker;
    private int liveness;
    private int interval;

    public CSBWorker(String conn) {
    	this.conn = conn;
    	ctx = new ZContext();
    }
    
	private Socket worker_socket() {
        Socket worker = ctx.createSocket(ZMQ.DEALER);
        id = CSBUtils.sockId();
        worker.setIdentity(id.getBytes());
        worker.connect(conn);
        
        ZFrame frame = new ZFrame(CSB.WORKER_READY);
        frame.send(worker, 0);
        
        LOG.info("worker created: " + id);
        return worker;
    }
	
	protected void doWork(String message) {}
	
	private IZLoopHandler proc = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	ZMsg msg = ZMsg.recvMsg(worker);
        	if (msg != null) {        	
        		if (msg.size() == 3) {
        			ZFrame address = msg.unwrap();
        		
        			doWork(new String(msg.pop().getData()));

        			ZMsg ack = ZMsg.newStringMsg(CSB.WORKER_ACK);
        			ack.wrap(address);
        			ack.send(worker);
        		
        			liveness = CSB.HEARTBEAT_LIVENESS;
        		} else if (msg.size() == 1) {
        			ZFrame frame = msg.getFirst();
        			if (CSB.WORKER_HEARTBEAT.equals(new String(frame.getData())))
        				liveness = CSB.HEARTBEAT_LIVENESS;
        			else {
        				LOG.warn(id + ": invalid message from router: "
        						+ new String(frame.getData()));
        			}
        			msg.destroy();
        		} else {
        			LOG.warn(id + ": malformed message from router: " 
        					+ new String(msg.getFirst().getData()));
        		}
        	
        		interval = CSB.RECONN_INTERVAL_INIT;
        	} else {
        		LOG.error(id + ": receiving interrupted.");
        	}
            return 0;
        }
	};
	
	private IZLoopHandler reconn = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	if (--liveness == 0) {
            	LOG.error(id + ": heartbeat failure, reconnecting in " + interval + " msec");
            	try {
            		Thread.sleep(interval);
            	} catch (InterruptedException e) { }
            	  
            	if (interval < CSB.RECONN_INTERVAL_MAX)
            		interval *= 2;
            	
        		loop.removePoller(new PollItem(worker, 0));
            	ctx.destroySocket(worker);
            	worker = worker_socket();
            	PollItem poller = new PollItem(worker, Poller.POLLIN);
            	loop.addPoller(poller, proc, null);

            	liveness = CSB.HEARTBEAT_LIVENESS;
        	}
        	
            ZFrame frame = new ZFrame(CSB.WORKER_HEARTBEAT);
            frame.send(worker, 0);

            return 0;
        }
	};
	
    public void run() {    	
        worker = worker_socket();
        liveness = CSB.HEARTBEAT_LIVENESS;
        interval = CSB.RECONN_INTERVAL_INIT;
        
        ZLoop loop = new ZLoop();
    	PollItem poller = new PollItem(worker, Poller.POLLIN);
    	loop.addPoller(poller, proc, null);
    	loop.addTimer(CSB.HEARTBEAT_INTERVAL, 0, reconn, null);        		
    	loop.start();
    	loop.destroy();
    	LOG.error(id + ": loop interrupted, exiting...");

    	ctx.destroy();
    }    
}

